/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.niotcp;

import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

import cgl.narada.transport.TransportDebugFlags;
import cgl.narada.transport.TransportHandler;
import cgl.narada.util.ByteUtilities;

/**
 * Listens for NIOTCP packets arriving at a socket and performs various
 * operations based on situations arising during communication. These received
 * packets are then encapsulated into a specialized data structure and forwarded
 * to the transport handler implenetation residing in the transport layer.
 * 
 * @author Shrideep Pallickara $Revision$ $Date$
 */

public class NIOTCPReceiverThread extends Thread implements TransportDebugFlags {

   private SocketChannel sc;

   private ByteBuffer buffer;

   private Selector readSelector;

   private int BUFFER_SIZE = 1024*64;

   private int packetTracker = 0;

   private boolean zeroLengthRead = false;

   private long readingDelayStart;

   private int intraPacketTracker = 0;

   private NIOTCPLink niotcpLink;

   private String moduleName = "NIOTCPReceiverThread: ";

   private TransportHandler transportHandler;

   private String linkId;

   private int numToRead = 0, readSoFar = 0;

   private byte[] dataRecvd;

   private int numToReadOffset = 0;

   private byte[] numToReadBytes;

   private int lastSizeRead = 0;

   private int tracker = 0;

   private int oldTracker = 0;

   private int receiveBufferSize = 12 * 8192;
   
   private boolean keepLooping = true;

   public NIOTCPReceiverThread(SocketChannel sc, NIOTCPLink niotcpLink) throws Exception {
      this.niotcpLink = niotcpLink;
      this.sc = sc;
      setPriority(MAX_PRIORITY);

      buffer = ByteBuffer.allocate(BUFFER_SIZE * 2);
      sc.configureBlocking(false);
      readSelector = Selector.open();
      sc.register(readSelector, SelectionKey.OP_READ);
      transportHandler = niotcpLink.getTransportHandler();
      if (transportHandler == null) {
         System.out.println(moduleName + "TransportHandler is null");
      }

      numToReadBytes = new byte[4];
      
      sc.socket().setReceiveBufferSize(receiveBufferSize);
   }

   public void setLinkId(String linkId) {
      if (linkId == null) {
         System.out.println(moduleName + "Specified Link ID is null");
      }
      this.linkId = linkId;
   }

   public void releaseResources() {
      if (NIOTCPReceiverThread_Debug)
         System.out.println(moduleName + "Closing " + sc);
      try {
         keepLooping = false;
         sc.close();
         /** Dealing with the closing part */
         readSelector.wakeup();
         readSelector.close();
      } catch (Exception ioe) {
         System.out.println(moduleName + "Problem while releasing resources" + ioe);
      }
   }

   public void run() {
      if (NIOTCPReceiverThread_Debug)
         System.out.println("NIOTCPReceiverThread listening for messages to " + sc);
      int dataLength;
      int packetTracker = 0;
      byte transportHeader;

      try {

         while (keepLooping) {
            if (NIOTCPReceiverThread_Debug) {
               System.out.println(moduleName + "tracker=" + tracker + ", readSoFar="
                                  + readSoFar + ", Remaining=" + (numToRead - readSoFar));
            }

            int status = readSelector.select();
            oldTracker = tracker;

            if (status == 0) {
               if (NIOTCPReceiverThread_Debug) {
                  System.out.println("\n \n Problem \n \n");
               }
               continue;
            }

            Iterator it = readSelector.selectedKeys().iterator();

            while (it.hasNext()) {
               SelectionKey key = (SelectionKey) it.next();

               if (key.isReadable()) {
                  readDataFromSocket();
               }

               it.remove(); /* remove key */             
            } /* end while */

            it = null;
            
            if (NIOTCPReceiverThread_Debug) {
               System.out
                         .println(moduleName + "Processed Socket data");
            }
         } /* end while(true) */
      } catch (Exception se) {
         //se.printStackTrace();
         System.out.println(moduleName + se);
      } /* end try-catch */

      /** Report loss of connection to the NIOTCP Link */
      niotcpLink.processLossOfSocketConnection(false);

      // if (NIOTCPReceiverThread_Debug)
      System.out.println("NIOTCPReceiver: Socket Connection to [" + 
                         niotcpLink.getLinkId() + "] broken");
   }

   public synchronized void readDataFromSocket() throws Exception {
      if (NIOTCPReceiverThread_Debug)
         System.out.println("Receiving ..");
      // SocketChannel sockChannel = (SocketChannel) key.channel();

      int count;
      buffer.clear();

      while ((count = sc.read(buffer)) > 0) {
         intraPacketTracker += count;
         tracker += count;

         if (NIOTCPReceiverThread_Debug) {
            System.out.println("Read ..." + tracker);
            System.out.println("intraPacketTracker=" + intraPacketTracker);
         }
         buffer.flip();

         if (buffer.remaining() != count) {
            System.out.println(moduleName + "Major Error buffer read mismatch!!");
         }

         processBuffer(count);

         /*
          * if ( (lastSizeRead+4) > buffer.capacity()) {
          * System.out.println(moduleName + "Reallocating buffer \n \n"); buffer =
          * ByteBuffer.allocateDirect(lastSizeRead+4); }
          */
      }

      if (count < 0) {
         System.out.println(moduleName + "Proceeding to close channel");
         sc.close();
         keepLooping = false;
      }

   }/* end readFromSocket */

   private void processBuffer(int count) {
      if (NIOTCPReceiverThread_Debug) {
         System.out.println("Processing buffer on link " + linkId + " ... size=" + count);
      }
      int advance = 0;
      if (numToRead == 0) {
         if (numToReadOffset == 0) {
            if (count >= 4) {
               numToRead = buffer.getInt();
               advance += 4;
               allocateDataReceived(numToRead);
            } else {
               // buffer.get(numToReadBytes, 0, count);
               buffer.get(numToReadBytes, numToReadOffset, count);
               numToReadOffset = count;
               System.out.println("numToReadOffset = " + numToReadOffset);
               advance += count;
            }
         } else {
            if ((numToReadOffset + count) >= 4) {
               System.out.println("numToReadOffset = " + numToReadOffset);
               buffer.get(numToReadBytes, numToReadOffset, (4 - numToReadOffset));
               numToRead = ByteUtilities.getInt(numToReadBytes);
               allocateDataReceived(numToRead);
               advance += (4 - numToReadOffset);
               numToReadOffset = 0;
            } else {
               buffer.get(numToReadBytes, numToReadOffset, count);
               numToRead = ByteUtilities.getInt(numToReadBytes);
               numToReadOffset += count;
               System.out.println("numToReadOffset = " + numToReadOffset);
               advance += count;
            }
         }
         /** numToRead Offset */
      }
      /** numToRead */

      if (advance == count) {
         return;
      }

      int availableToRead = count - advance;
      if (NIOTCPReceiverThread_Debug) {
         System.out.println("availableToRead=" + availableToRead + ", numToRead="
                            + numToRead + ", readSoFar=" + readSoFar + ", totalReceived="
                            + tracker);
      }
      if (readSoFar == 0) {
         if (availableToRead >= numToRead) {
            buffer.get(dataRecvd, 0, numToRead);
            advance += numToRead;
            if (NIOTCPReceiverThread_Debug) {
               System.out.println("Reporting [" + numToRead
                                  + "] to TransportHandler in one read");
            }
            transportHandler.dataReceived(dataRecvd, linkId);
            intraPacketTracker -= (numToRead + 4);
            numToRead = 0;
            readSoFar = 0;

            if ((count - advance) > 0)
               processBuffer(count - advance);
         } else {
            buffer.get(dataRecvd, 0, availableToRead);
            readSoFar = availableToRead;
            return;
         }
      } else {
         if ((availableToRead + readSoFar) >= numToRead) {
            buffer.get(dataRecvd, readSoFar, (numToRead - readSoFar));
            advance += (numToRead - readSoFar);
            readSoFar += (numToRead - readSoFar);
            if (NIOTCPReceiverThread_Debug) {
               System.out.println("Reporting [" + numToRead
                                  + "] to TransportHandler greater reads");
            }
            transportHandler.dataReceived(dataRecvd, linkId);
            intraPacketTracker -= (numToRead + 4);
            numToRead = 0;
            readSoFar = 0;
            if ((count - advance) > 0)
               processBuffer(count - advance);
         } else {
            buffer.get(dataRecvd, readSoFar, availableToRead);
            readSoFar += availableToRead;
         }
      }

   }

   private void allocateDataReceived(int size) {
      if (size > 100000) {
         //System.out.println("Number of bytes to read = [" + size + "] on link " + linkId);
      }

      
      /**if (dataRecvd != null) { 
          if (dataRecvd.length == size) 
              return; 
      }*/
      
      dataRecvd = new byte[size];
   }

}
